package com.tensai.study.java.newconcept.typereactiveprogram.reactor.exception;

import org.junit.Test;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;

/**
 * 异常处理测试类
 *
 * @author Tensai
 */
public class ExceptionTest {

	/**
	 * 先举一个例子，我们创建一个Flux，在这个Flux中，我们产生一个异常，看看是什么情况
	 */
	@Test
	public void demo() {
		Flux<String> flux2 = Flux.just(1, 2, 0)
				.map(i -> "100 / " + i + " = " + (100 / i));
		flux2.subscribe(System.out::println);

	}

	/**
	 * 有两种方式，第一种方式就是我们之前文章讲过的，在subscribe的时候指定onError方法
	 * Fallback Method
	 */
	@Test
	public void onErrorResumeTest() {
		Flux<String> flux2 = Flux.just(1, 2, 0)
				.map(i -> "100 / " + i + " = " + (100 / i));

		flux2.subscribe(System.out::println,
				error -> System.err.println("Error: " + error));
	}

	/**
	 * 除了在subscribe中进行处理，我们还可以在publish的时候，就指定异常的处理模式，这就是我们要介绍的第二种方法
	 * Static Fallback Value的意思是，在遇到异常的时候会fallback到一个静态的默认值
	 */
	@Test
	public void onErrorReturnTest() {
		Flux<String> flux = Flux.just(1, 2, 0)
				.map(i -> "100 / " + i + " = " + (100 / i))
				.onErrorReturn("Divided by zero :(");
		flux.subscribe(System.out::println);

	}

	/**
	 * Dynamic Fallback Value
	 * 所谓的动态Fallback Value就是根据你抛出的异常进行判断，通过定位不同的Error从而fallback到不同的值：
	 */
	@Test
	public void useDynamicFallback() {
		Flux<String> flux = Flux.just(1, 2, 0)
				.map(i -> "100 / " + i + " = " + (100 / i))
				.onErrorResume(error -> Mono.just(
						MyWrapper.fromError(error)));
		flux.subscribe(System.out::println);
	}

	@Test
	public void onErrorMapTest() {
		Flux<String> flux2 = Flux.just(1, 2, 0)
				.map(i -> "100 / " + i + " = " + (100 / i))
				.onErrorMap(error -> new RuntimeException("oops, ArithmeticException!", error));
		flux2.subscribe(System.out::println);
	}

	/**
	 * 有时候你只是想记录一下异常信息，并不想破坏原来的React结构，那么可以试着使用doOnError。
	 */
	@Test
	public void doOnErrorTest() {
		Flux<String> flux = Flux.just(1, 2, 0, 6)
				.map(i -> "100 / " + i + " = " + (100 / i))
				.doOnError(error -> System.out.println("we got the error: " + error));
		flux.subscribe(System.out::println);
	}

	@Test
	public void doFinallyTest() {
		LongAdder statsCancel = new LongAdder();
		Flux<String> flux =
				Flux.just("foo", "bar")
						.doOnSubscribe(System.out::println)
						.doFinally(type -> {
							if (type == SignalType.CANCEL) {
								statsCancel.increment();
							}
						})
						.take(1);
		flux.subscribe();

	}

	/**
	 * 可以看到using支持三个参数，resourceSupplier是一个生成器，用来在subscribe的时候生成要发送的resource对象。
	 * <p>
	 * sourceSupplier是一个生成Publisher的工厂，接收resourceSupplier传过来的resource，然后生成Publisher对象。
	 * <p>
	 * resourceCleanup用来对resource进行收尾操作。
	 */
	@Test
	public void usingTest() {
		AtomicBoolean isDisposed = new AtomicBoolean();
		Disposable disposableInstance = new Disposable() {
			@Override
			public void dispose() {
				isDisposed.set(true);
			}
			@Override
			public String toString() {
				return "DISPOSABLE";
			}
		};

		Flux<String> flux =
				Flux.using(
						() -> disposableInstance,
						disposable -> Flux.just(disposable.toString()),
						Disposable::dispose);
		flux.subscribe(System.out::println);
	}

	/**
	 * 异常重试机制
	 * 重新开始，而不是从出错的地方开始
	 */
	@Test
	public void retryTest() {
		Flux.interval(Duration.ofMillis(250))
				.map(input -> {
					if (input < 3) {
						return "tick " + input;
					}
					throw new RuntimeException("boom");
				})
				.retry(1)
				.elapsed()
				.subscribe(System.out::println, System.err::println);

		try {
			Thread.sleep(2100);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}


	public static class MyWrapper {
		public static String fromError(Throwable error) {
			return "That is a new Error: " + error.getMessage();
		}
	}

}
